package com.laowang.wordcount;


import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;

import javax.lang.model.type.ArrayType;
import javax.xml.transform.Source;
import java.util.ArrayList;
import java.util.List;

public class WordCountSQL {

    public static void main(String[] args) throws Exception{

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);

        String words = "hello flink hello imooc";

        String[] split = words.split("\\W+");
        List<WC> list = new ArrayList();

        for(String word : split){

            WC wc = new WC(word,1);
            list.add(wc);
        }
        DataSet<WC> input = env.fromCollection(list);
        tEnv.registerDataSet("WordCount",input,"word, frequency");
        //String sql = "select word, sum(frequency) as frequency from WordCount GROUP BY word";

        //比较函数 (等于/like)
        //String sql = "select word, sum(frequency) as frequency from WordCount where word = 'hello' GROUP BY word";
        // String sql = "select word, sum(frequency) as frequency from WordCount where word LIKE '%i%' GROUP BY word";
        //算术函数 (+/-)
         //String sql = "select word, sum(frequency) - 1 as frequency from WordCount GROUP BY word";


        //字符串处理函数(UPPER/LOWER/SUBSTRING)
        //String sql = "select UPPER(word) as word, sum(frequency) as frequency from WordCount GROUP BY word";
        //String sql = "select SUBSTRING(word,2) as word, sum(frequency) as frequency from WordCount GROUP BY word";

        //其他函数(MD5)
        String sql = "select MD5(word) as word, sum(frequency) as frequency from WordCount GROUP BY word";

        Table table = tEnv.sqlQuery(sql);

        DataSet<WC> wcDataSet = tEnv.toDataSet(table, WC.class);
        wcDataSet.print();

    }//main



    public static class WC {
        public String word;//hello flink hello imooc
        public long frequency;//1

        // public constructor to make it a Flink POJO
        public WC() {}

        public WC(String word, long frequency) {
            this.word = word;
            this.frequency = frequency;
        }

        @Override
        public String toString() {
            return  word + ", " + frequency;
        }
    }

}//
